package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.RegistryConsumerGroupEvent;
import org.tao.lightningmq.broker.model.ConsumeQueueOffsetModel;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.model.QueueModel;
import org.tao.lightningmq.broker.rebalance.ConsumerInstance;
import org.tao.lightningmq.common.dto.RegistryConsumerGroupReqDTO;
import org.tao.lightningmq.common.dto.RegistryConsumerGroupRespDTO;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RegistryConsumerGroupListener implements Listener<RegistryConsumerGroupEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RegistryConsumerGroupListener.class);

    @Override
    public void onReceive(RegistryConsumerGroupEvent event) throws Exception {
        // 先初始化消费者组的消费进度
        RegistryConsumerGroupReqDTO reqDTO = event.getReqDTO();
        String consumeGroup = reqDTO.getConsumeGroup();
        String topic = reqDTO.getTopic();

        ConsumeQueueOffsetModel.OffsetTable offsetTable = CommonCache.getConsumeQueueOffsetModel().getOffsetTable();
        ConsumeQueueOffsetModel.ConsumerGroupDetail consumerGroupDetail = offsetTable.getTopicConsumerGroupDetail().get(topic);
        if (consumerGroupDetail == null) {
            consumerGroupDetail = new ConsumeQueueOffsetModel.ConsumerGroupDetail();
            offsetTable.getTopicConsumerGroupDetail().put(topic, consumerGroupDetail);
        }

        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        Map<String, Map<String, String>> consumeGroupOffsetMap = consumerGroupDetail.getConsumerGroupDetailMap();
        Map<String, String> queueOffsetDetailMap = consumeGroupOffsetMap.get(consumeGroup);
        List<QueueModel> queueList = mqTopicModel.getQueueList();
        if (queueOffsetDetailMap == null) {
            queueOffsetDetailMap = new HashMap<>();
            for (QueueModel queueModel : queueList) {
                queueOffsetDetailMap.put(String.valueOf(queueModel.getId()), "00000000#0");
            }
            consumeGroupOffsetMap.put(consumeGroup, queueOffsetDetailMap);
        }

        LOGGER.info("消费者组: {} 的消费进度初始化完毕!", consumeGroup);

        // 再把对应的消费者添加到消费池中
        String currentReqId = reqDTO.getConsumerReqId();
        ConsumerInstance consumerInstance = new ConsumerInstance();
        consumerInstance.setIp(reqDTO.getIp());
        consumerInstance.setPort(reqDTO.getPort());
        consumerInstance.setConsumerReqId(currentReqId);
        consumerInstance.setTopic(topic);
        consumerInstance.setConsumeGroup(consumeGroup);
        consumerInstance.setBatchSize(reqDTO.getBatchSize());

        //加入到消费池中
        CommonCache.getConsumerInstancePool().addInstancePool(consumerInstance);

        LOGGER.info("消费者: {} 加入到消费池中!", reqDTO.getConsumerReqId());

        RegistryConsumerGroupRespDTO respDTO = new RegistryConsumerGroupRespDTO(true);
        respDTO.setMsgId(event.getMsgId());
        TcpMsg respMsg = new TcpMsg(BrokerResponseCode.REGISTRY_CONSUMER_GROUP_RESP.getCode(),
                JSON.toJSONBytes(respDTO));
        event.getContext().writeAndFlush(respMsg);

    }
}
